package com.xpec.c4.service.module.async;

import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.xpec.c4.game.time.TimeSys;
import com.xpec.c4.service.module.BasicServiceModule;
import com.xpec.c4.service.module.RecordedQueue;
import com.xpec.c4.service.module.ThreadSection;
import com.xpec.c4.service.module.task.TaskAsyncCommand;


/**
 * 異步命令處理
 */
public class AsyncCommandProcedure implements Runnable{

	private static final Log log = LogFactory.getLog(AsyncCommandProcedure.class);
	
	/**
	 * 目前的 procedure 的 queue id<br>
	 * 屬於第幾條執行緒, 0 base
	 */
	private int queueId;

	/**
	 * 
	 */
	private Queue<TaskAsyncCommand> queue;
	
	private AtomicInteger taskCounter = new AtomicInteger();
	
	public int getQueueSize()
	{
//		return queue.size();
		return taskCounter.get();
	}
	
	public Map<Class, AtomicInteger> getCountingMap()
	{
		if(queue instanceof RecordedQueue) 
		{
			return ((RecordedQueue)queue).getCountingMap();
		}
		else
		{
			return new HashMap<Class, AtomicInteger>();
		}
	}
	
	/**
	 * 
	 */
	private boolean aboutToClose = false;
	
	/**
	 * 
	 */
	private Object waitObj = new Object();
	
	/**
	 * 
	 */
	protected Thread threadInfo = null;
	
	/**
	 * 唤醒线程闲置
	 */
	protected void threadNotify(){		
		synchronized(waitObj){
			waitObj.notify();
		}		
	}

	/**
	 * 記下module
	 */
	private BasicServiceModule serverModule;
	
	/**
	 * @param queueId
	 *            這個 procedure 的 queueId, 參考: {@link #queueId}
	 * @param basicServiceModule
	 */
	public AsyncCommandProcedure(int queueId, BasicServiceModule basicServiceModule) {
		this(queueId, basicServiceModule, false);
	}

	/**
	 * @param queueId
	 *            這個 procedure 的 queueId, 參考: {@link #queueId}
	 * @param basicServiceModule
	 */
	public AsyncCommandProcedure(int queueId, BasicServiceModule basicServiceModule, boolean useRecordedQueue) {
		
		queue = useRecordedQueue ? new RecordedQueue<TaskAsyncCommand>(new ConcurrentLinkedQueue<TaskAsyncCommand>())
				: new ConcurrentLinkedQueue<TaskAsyncCommand>();
		
		this.serverModule = basicServiceModule;

		this.threadInfo = new Thread(this);

		this.queueId = queueId;
		
		String s = this.getClass().getSimpleName();
		String threadName = "c4_" + serverModule.getModuleName() + "_q" + queueId + "_" + s + "_"
				+ threadInfo.getId();
		
		threadInfo.setName(threadName);
	}
	
	/**
	 * 启动线程
	 */
	public boolean startup(){		
		threadInfo.start();
		return true;
	}	

	/**
	 * 
	 */
	public void shutdown() {
		aboutToClose = true;
		threadInfo = null;
	}

	
	/**
	 * 
	 * @param obj
	 * @return
	 */
	public boolean addAsyncCommand(TaskAsyncCommand obj){
		boolean result = queue.add(obj);
		//加進去queue就+1
		taskCounter.incrementAndGet();
		threadNotify();
		return result;
	}

	@Override
	public void run() {
		//long startTime, endTime, operatorTime;
		//String logTitle;
		// 一樣 把queue內的obj取出來執行		
		// 這邊會block, 要自己實現欲調用的邏輯
		
		while(true){

			try {

					while(!queue.isEmpty()){
						
						TaskAsyncCommand obj = queue.poll();
						//拿出來就扣掉
						taskCounter.decrementAndGet();
	
						if (obj != null) {
							//startTime = System.currentTimeMillis();
							
							// 結果 成功或失敗execute自行記錄在obj內
							
							//modified by Miriam_Chen on 2013/4/10 : 新增DCAsyncCommandWithSyncReturn類別處理,如果不是DCAsyncCommandWithSyncReturn,就維持原程式碼
	//						obj.asyncExecute();
	//						
	//						// 将处理完成的命令丢回module
	//						serverModule.addTaskObject(obj);
//							if (obj instanceof DCAsyncCommandWithSyncReturn) {
//								obj.asyncExecute();
//								//不是DCAsyncCommandWithSyncReturn,就維持原程式碼
//							} 
//							else {

								long execBeginTime = TimeSys.currentTimeMills();
								
								obj.asyncExecute();
								
								long spendTime = TimeSys.currentTimeMills() - execBeginTime;
								if(spendTime > ThreadSection.TASK_EXECUTE_WARN_TIME){
									if (log.isErrorEnabled()) {
										log.error("q" + queueId + " task=" + queue.size() + " "
												+ obj.getRealTaskName() + ".asyncExecute() spend time too long "
												+ spendTime + " ms");
									}
								}
								
								// 将处理完成的命令丢回module
								serverModule.addTaskObject(obj);
//							}
							//end miriam
							/*endTime = System.currentTimeMillis();
							operatorTime = endTime - startTime;
							logTitle = 	obj.getClass().getName() + ": AsyncExecute";
							serverModule.updataExecuteCount(logTitle, operatorTime);*/
						}
					}

					// 没消息时让出线程时间
					synchronized (waitObj) {
						if(aboutToClose)
						{
							terminated = true;
							return;
						}
						// 1秒苏醒一次, waiting期间也可被唤醒
						waitObj.wait(1000);
					}
				
			} catch (Throwable e) {
				// 因為是線程就全catch咯...
				// e.printStackTrace();
				log.error("error!", e);
			}

		}		
		
	}
	
	private volatile boolean terminated = false; 
	
	public boolean isTerminated() {
		return terminated;
	}

	public int getQueueId() {
		return queueId;
	}	
	
}
